#include <iostream>
#include <cstring>

#include "service.h"
#include "sunnet.h"
#include "luaapi.h"

using namespace std;

service::service()
{
    // 初始化锁
    // PTHREAD_PROCESS_PRIVATE, 表示进程私有, 进程内线程共享
    pthread_spin_init(&this->msg_queue_lock, PTHREAD_PROCESS_PRIVATE);
    pthread_spin_init(&this->is_in_global_queue_lock, PTHREAD_PROCESS_PRIVATE);
}

service::~service()
{
    // 销毁 spinlock
    pthread_spin_destroy(&this->msg_queue_lock);
    pthread_spin_destroy(&this->is_in_global_queue_lock);
}

void service::push_msg(shared_ptr<base_msg> msg)
{
    pthread_spin_lock(&this->msg_queue_lock);
    {
        msg_queue.push(msg);
    }
    pthread_spin_unlock(&this->msg_queue_lock);
}

shared_ptr<base_msg> service::pop_msg()
{
    shared_ptr<base_msg> msg = nullptr;
    pthread_spin_lock(&this->msg_queue_lock);
    {
        if (!this->msg_queue.empty())
        {
            msg = this->msg_queue.front();
            this->msg_queue.pop();
        }
    }
    pthread_spin_unlock(&this->msg_queue_lock);
    return msg;
}

void service::on_init()
{
    cout << "service " << this->id << " on init" << endl;

    // 测试, 开始监听
    // sunnet::inst->Listen(8000, this->id);

    // init lua state.
    luastate = luaL_newstate();
    luaL_openlibs(luastate); // open all previous libraries.
    // register sunnet utilits
    luaapi::Register(luastate);
    // run the main lua file.
    string filename = "../service/" + *type + "/init.lua"; // 创建的时候, type 被赋值为 main. refer to main.cpp
    int isok = luaL_dofile(luastate, filename.data());     // ??? c++ 特有的东西
    if (isok == 1)
        cout << "run the main lua file fail: " << lua_tostring(luastate, -1) << endl;

    // call init function of lua
    /**
     * ???
     * 访问 lua 语言里面的方法或者变量的方式??? 必须是全局的吗?
     * special, 如何访问一个 table 中的一个方法 ???
     */
    lua_getglobal(luastate, "on_init");
    lua_pushinteger(luastate, id);
    isok = lua_pcall(luastate, 1, 0, 0); // int lua_pcall (lua_State *L, int nargs, int nresults, int msgh); // msgh->msghandler
    if (isok != 0)
        cout << "call lua on_init fail" << lua_tostring(luastate, -1) << endl;
}

void service::on_msg(shared_ptr<base_msg> msg)
{
    if (msg->type == base_msg::TYPE::service)
    {
        auto m = dynamic_pointer_cast<service_msg>(msg);
        on_service_msg(m);
    }
    else if (msg->type == base_msg::TYPE::socket_accept)
    {
        auto m = dynamic_pointer_cast<socket_accept_msg>(msg);
        on_accept_msg(m);
    }
    else if (msg->type == base_msg::TYPE::socket_rw)
    {
        auto m = dynamic_pointer_cast<socket_rw>(msg);
        on_rw_msg(m);
    }
    else
        cout << this->id << " msg type error, " << msg->type << endl;
}

void service::on_exit()
{
    cout << this->id << "on exit" << endl;

    lua_getglobal(luastate, "on_exit");
    int isok = lua_pcall(luastate, 0, 0, 0);
    if (isok != 0)
        cout << "call call lua on_init fail " << lua_tostring(luastate, -1) << endl;

    // close the lua state
    lua_close(luastate);
}

// 处理消息
bool service::process_msg()
{
    shared_ptr<base_msg> msg = this->pop_msg();
    if (msg)
    {
        on_msg(msg);
        return true;
    }
    else
        return false;
}

// 处理多条消息
void service::process_msgs(int max)
{
    for (int i = 0; i < max; i++)
    {
        bool succ = this->process_msg();
        if (!succ)
            break;
    }
}

void service::set_in_global(bool is_in)
{
    pthread_spin_lock(&this->is_in_global_queue_lock);
    {
        this->is_in_global_queue = is_in;
    }
    pthread_spin_unlock(&this->is_in_global_queue_lock);
}

void service::on_service_msg(shared_ptr<service_msg> msg)
{
    cout << "on service msg " << endl;
    // 调用 lua 层
    lua_getglobal(luastate, "onservicemsg");
    lua_pushinteger(luastate, msg->source);
    lua_pushlstring(luastate, msg->buff.get(), msg->size);
    int isok = lua_pcall(luastate, 2, 0, 0);
    if (isok != 0)
        cout << "call lua onservicemsg fail" << lua_tostring(luastate, -1) << endl;
}

void service::on_accept_msg(shared_ptr<socket_accept_msg> msg)
{
    cout << "on accept msg" << endl;
    // 调用 lua 层
    lua_State *l = this->luastate;
    lua_getglobal(l, "on_accept_msg");
    lua_pushinteger(l, msg->client_fd);
    int isok = lua_pcall(l, 1, 0, 0);
    if (isok != LUA_OK)
        cout << "call lua on_accept_msg fail: " << lua_tostring(l, -1) << endl;
}

void service::on_rw_msg(shared_ptr<socket_rw> msg)
{
    int fd = msg->fd;
    if (msg->is_read)
    {
        const int BUFFSIZE = 512;
        char buff[BUFFSIZE];
        int len = 0;
        do
        {
            // len = read(fd, &buff, BUFFSIZE);
            len = read(fd, buff, BUFFSIZE); // 不加 & 符号更清晰, 数组名会自动变为指针
            if (len > 0)
            {
                this->on_socket_date(fd, buff, len);
            }
        } while (len == BUFFSIZE);

        if (len <= 0 && errno != EAGAIN) // errno 是线程唯一的. TLS(thread local storage)
        {
            if (sunnet::inst->get_conn(fd))
            {
                this->on_socket_close(fd);
                sunnet::inst->close_conn(fd);
            }
        }
    }

    if (msg->is_write)
    {
        if (sunnet::inst->get_conn(fd))
            this->on_socket_write(fd);
    }
}

void service::on_socket_date(int fd, const char *buff, int len)
{
    // cout << "on socket data " << fd << " len: " << len << " buff: " << buff;
    // char write_buff[] = "copy that\n";
    // write(fd, write_buff, sizeof(write_buff));

    // 调用 lua 层的函数
    lua_State *l = this->luastate;
    lua_getglobal(l, "on_socket_date");
    lua_pushinteger(l, fd);
    lua_pushlstring(l, buff, len);
    int isok = lua_pcall(l, 2, 0, 0); // lua_call 如果遇到问题会通过 longjmp 传播, 如果没有处理会导致 undefined behaviour or exit.
    if (isok != LUA_OK)
        // lua 5.4 允许抛出任意的错误对象. string, integer, table, ...
        // 如果是报错, lua 默认抛出的是 string.
        cout << "call lua onsocketdata fail: " << lua_tostring(l, -1) << endl;
}

/**
 * 测试发送大量数据
 * 一次性是写不完的,
 * 1. 设置缓冲区的大小, setsockopt, ...
 * 2. 自写缓冲区,
 */
// void service::on_socket_date(int fd, const char *buff, int len)
// {
//     cout << "on socket data" << fd << " buff: " << buff << endl;
//     char write_buff[4200000] = "l";
//     write_buff[4200000 - 2] = 'y';
//     write_buff[4200000 - 1] = '\0';
//     int r = write(fd, write_buff, sizeof(write_buff));
//     cout << "on socket data, r = " << r << endl;
// }

/**
 * // 测试 linux 的一个坑, 在等待的 15 s 期间, 如果客户端退出了, 进程会直接崩溃
 * // 原因是向一个已经关闭了的 tcp peer 写数据的话, 会回复一个 RST 信号
 * // 默认情况下, 向一个收到复位信号的 socket 写数据的时候, 操作系统会发送 SIGPIPE 信号,
 * // 默认关闭进程.
 * // 处理办法是忽略该信号, signal(SIGPIPE, SIG_IGN);
 */
// void service::on_socket_date(int fd, const char *buff, int len)
// {
//     cout << "on socket data" << fd << " buff: " << buff << endl;
//     // echo
//     char write_buff[] = "lpy";
//     write(fd, write_buff, sizeof(write_buff));
//     // sleep 15 seconds
//     usleep(15 * 1000000);
//     char writebuff2[] = "something2";
//     int r = write(fd, writebuff2, sizeof(writebuff2));
//     cout << "write 2 :" << r << strerror(errno) << endl;
//     // sleep 1 seconds
//     usleep(1 * 1000000);
//     char writebuff3[] = "sth3";
//     r = write(fd, writebuff3, sizeof(writebuff3));
//     cout << "write 3 :" << r << strerror(errno) << endl;
// }

void service::on_socket_write(int fd)
{
    cout << "on socket writeable " << fd << endl;
}

void service::on_socket_close(int fd)
{
    cout << "on socket close " << fd << endl;
    lua_State *l = this->luastate;
    lua_getglobal(l, "on_socket_close"); // 如果对应的方法不存在, 则将 nil 值压入栈顶
    // int num = lua_gettop(l);
    // cout << "The number of the stack is " << num << endl; // is 1
    lua_pushinteger(l, fd);
    int r = lua_pcall(l, 1, 0, 0);
    if (r != LUA_OK)
        cout << "call on_socket_close fail: " << lua_tostring(l, -1) << endl;
}